package com.shujia.flink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{ProcessingTimeSessionWindows, SlidingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object StreamWindow {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //通过socket创建dataStream
    val ds = env.socketTextStream("node1", 8888)


    val countDS = ds
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(0) //不能以数组作为key   如果使用自定义对象，需要实现hashcode方法

      //.timeWindow(Time.seconds(5)) //滚动窗口  5s一个

      .timeWindow(Time.seconds(5), Time.seconds(1)) //滑动窗口

      //SlidingProcessingTimeWindows   处理时间的滑动窗口
      //      .window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))

      .sum(1)

    //countDS.print()


    ds
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(0)
      //.countWindow(10) //每个key累10条数据触发计算    滚动窗口
      .countWindow(10, 2) //滑动窗口
      .sum(1)
    //.print()


    ds
      .flatMap(_.split(","))
      .map((_, 1))
      .keyBy(0)
      //会话窗口
      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))) //如果5s没有数据过来触发windows
      .sum(1)
    .print()


    env.execute()

  }
}
